package com.atguigu.gmall.realtime.dws.app;

import com.atguigu.gmall.realtime.common.base.BaseSQLApp;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.SQLUtil;
import com.atguigu.gmall.realtime.dws.function.KeyWordUDTF;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwsTrafficSourceKeywordPageViewWindow extends BaseSQLApp {
    public static void main(String[] args) {
        new DwsTrafficSourceKeywordPageViewWindow().start(10021, 4,"dws_traffic_source_keyword_page_view_window");
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        // TODO 1.注册自定义函数到表的执行环境中
        tableEnv.createFunction("ik", KeyWordUDTF.class);
        // TODO 2.从kafka的页面日志中读取数据创建状态表 并指定Watermark的生成策略以及提及事件时间字段
        tableEnv.executeSql("CREATE TABLE page_log (\n" +
                "  common map<string,string>,\n" +
                "  page map<string,string>,\n" +
                "  ts BIGINT,\n" +
                "  et as TO_TIMESTAMP_LTZ(ts, 3),\n" +
                "  watermark FOR et as et\n" +
                ")" + SQLUtil.getKafkaDDL(Constant.TOPIC_DWD_TRAFFIC_PAGE, "dws_traffic_source_keyword_page_view_window"));
        //tableEnv.executeSql("select * from page_log").print();
        // TODO 3.过滤搜索行为
        Table searchTable = tableEnv.sqlQuery("select\n" +
                " page['item'] fullword,\n" +
                " et\n" +
                " from page_log\n" +
                " where page['last_page_id'] = 'search' and page['item_type'] = 'keyword' and page['item'] is not null");
        tableEnv.createTemporaryView("search_table", searchTable);
        //tableEnv.executeSql("select * from search_table");
        // TODO 4.使用自定义函数进行分词 并与原表的的字段进行关联
        Table splitTable = tableEnv.sqlQuery("select\n" +
                "    keyword,\n" +
                "    et\n" +
                "from search_table, lateral table(ik(fullword)) t(keyword)");
        tableEnv.createTemporaryView("split_table", splitTable);
        tableEnv.executeSql("select * from split_table").print();
        // TODO 5.分组，开窗 聚合
        Table resTable = tableEnv.sqlQuery("select\n" +
                "    DATE_FORMAT(window_start, 'yyyy-MM-dd HH:mm:ss') stt,\n" +
                "    DATE_FORMAT(window_end, 'yyyy-MM-dd HH:mm:ss') edt,\n" +
                "    DATE_FORMAT(window_start, 'yyyy-MM-dd') cur_date,\n" +
                "    keyword,\n" +
                "    count(*) keyword_count\n" +
                "  from table(\n" +
                "    tumble(table split_table, descriptor(et), interval '10' second))\n" +
                "  group by keyword, window_start, window_end");
        //resTable.execute().print();
        // TODO 6.将聚合结果写入到doris
        //6.1 创建动态表和要写入的Doris表进行映射
//        tableEnv.executeSql("create table dws_traffic_source_keyword_page_view_window(\n" +
//                "    stt string,\n" +
//                "    edt string,\n" +
//                "    cur_date string,\n" +
//                "    keyword string,\n" +
//                "    keyword_count bigint\n" +
//                ")" + SQLUtil.getDorisDDL("dws_traffic_source_keyword_page_view_window"));
//        //6.2 写入
//        resTable.executeInsert("dws_traffic_source_keyword_page_view_window");
    }
}

